package com.lixixi.tingshu.user.service.impl;

import cn.hutool.core.bean.BeanUtil;
import com.alibaba.fastjson.JSONObject;
import com.lixixi.tingshu.common.constant.KafkaConstant;
import com.lixixi.tingshu.common.constant.RedisConstant;
import com.lixixi.tingshu.common.constant.SystemConstant;
import com.lixixi.tingshu.common.util.AuthContextHolder;
import com.lixixi.tingshu.common.util.MongoUtil;
import com.lixixi.tingshu.model.user.UserListenProcess;
import com.lixixi.tingshu.user.service.UserListenProcessService;
import com.lixixi.tingshu.vo.album.AlbumStatMqVo;
import com.lixixi.tingshu.vo.album.TrackStatMqVo;
import com.lixixi.tingshu.vo.user.UserListenProcessVo;
import lombok.extern.slf4j.Slf4j;
import org.joda.time.DateTime;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Sort;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import java.math.BigDecimal;
import java.time.Duration;
import java.time.LocalDateTime;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

@Slf4j
@Service
public class UserListenProcessServiceImpl implements UserListenProcessService {
	@Autowired
	MongoTemplate mongoTemplate;

	@Autowired
	RedisTemplate redisTemplate;

	@Autowired
	KafkaTemplate kafkaTemplate;

	@Override
	public BigDecimal getTrackBreakSecond(Long trackId) {
		//根据专辑id查询记录
		Query query = Query.query(Criteria
				.where("userId")
				.is(AuthContextHolder.getUserId())
				.and("trackId")
				.is(trackId));
		UserListenProcess listenProcess = mongoTemplate.findOne(query, UserListenProcess.class
				, MongoUtil.getCollectionName(MongoUtil.MongoCollectionEnum.USER_LISTEN_PROCESS
						, AuthContextHolder.getUserId()));
		if (BeanUtil.isEmpty(listenProcess))
		{
			//无上次播放记录则返回进度 : 0
			return BigDecimal.valueOf(0);
		}
		return listenProcess.getBreakSecond();
	}

	@Override
	public void updateListenProcess(UserListenProcessVo userListenProcessVo) {
		//获取专辑id
		Long albumId = userListenProcessVo.getAlbumId();
		//获取声音id
		Long trackId = userListenProcessVo.getTrackId();
		//获取用户id
		Long userId = AuthContextHolder.getUserId();
		//通过声音id以及用户id查询当前声音记录
		Query query = Query.query(Criteria
				.where("userId")
				.is(AuthContextHolder.getUserId())
				.and("trackId")
				.is(trackId));
		//接收查询数据
		UserListenProcess listenProcess = mongoTemplate.findOne(query, UserListenProcess.class
				, MongoUtil.getCollectionName(MongoUtil.MongoCollectionEnum.USER_LISTEN_PROCESS
						, AuthContextHolder.getUserId()));
		//记录为空则创建并保存该记录
		if (BeanUtil.isEmpty(listenProcess))
		{
			//为空创建对象并填充属性
			listenProcess = new UserListenProcess();
			listenProcess.setUserId(userId);
			listenProcess.setTrackId(trackId);
			listenProcess.setAlbumId(albumId);
			listenProcess.setBreakSecond(userListenProcessVo.getBreakSecond());
			listenProcess.setCreateTime(new Date());
			listenProcess.setUpdateTime(new Date());
			mongoTemplate.save(listenProcess, MongoUtil
					.getCollectionName(MongoUtil.MongoCollectionEnum.USER_LISTEN_PROCESS
							, AuthContextHolder.getUserId()));
		} else
		{
			//存在则更新其声音播放进度和更新时间,并保存
			listenProcess.setBreakSecond(userListenProcessVo.getBreakSecond());
			listenProcess.setUpdateTime(new Date());
			mongoTemplate.save(listenProcess, MongoUtil
					.getCollectionName(MongoUtil.MongoCollectionEnum.USER_LISTEN_PROCESS
							, AuthContextHolder.getUserId()));
		}
		//附带功能
		//发送消息,更新声音和专辑的 '统计类型',每个用户一天只能记录一次
		//使用位图记录当前用户今日是否播放改声音
		String bitMapKey = RedisConstant.BIT_MAP_TRACK + ":" + new DateTime().toString("yyyyMMdd") + ":" + userId;
		//获取该用户的位图
		Boolean trackExist = redisTemplate.opsForValue().getBit(bitMapKey, trackId);

		if (Boolean.TRUE.equals(trackExist))
		{
			//今日已播放过该声音,不进行更新
			log.info("用户{}今日已播放声音{},直接返回", userId, trackId);
			return;
		}
		//初次播放,对其进行加锁
		Boolean absent = redisTemplate.opsForValue().setIfAbsent("SendMessage-updateStat:" + userId + ":" + trackId
				, 1
				, RedisConstant.CACHE_TEMPORARY_TIMEOUT
				, TimeUnit.SECONDS);
		//获取锁成功则进行发送消息操作
		if (absent)
		{
			redisTemplate.opsForValue().setBit(bitMapKey, trackId, Boolean.TRUE);
			//设置bitMap过期时间,期限为1天
			LocalDateTime localDateTime = java.time.LocalDateTime.now().plusDays(1);
			long between = Duration.between(LocalDateTime.now(), localDateTime).getSeconds();
			redisTemplate.expire(bitMapKey, between, TimeUnit.SECONDS);
			//构建发送消息载体
			TrackStatMqVo trackStatMqVo = new TrackStatMqVo();
			//设立业务编号,保证消息幂等性
			trackStatMqVo.setBusinessNo(UUID.randomUUID().toString().replace("-", ""));
			trackStatMqVo.setStatType(SystemConstant.TRACK_STAT_PLAY);
			trackStatMqVo.setAlbumId(albumId);
			trackStatMqVo.setTrackId(trackId);
			trackStatMqVo.setCount(1);
			//声音统计类型
			kafkaTemplate.send(KafkaConstant.QUEUE_TRACK_STAT_UPDATE, JSONObject.toJSONString(trackStatMqVo));

			//构建发送消息载体
			AlbumStatMqVo albumStatMqVo = new AlbumStatMqVo();
			//设立业务编号,保证消息幂等性
			albumStatMqVo.setBusinessNo(UUID.randomUUID().toString().replace("-", ""));
			albumStatMqVo.setAlbumId(albumId);
			albumStatMqVo.setStatType(SystemConstant.ALBUM_STAT_PLAY);
			albumStatMqVo.setCount(1);
			//专辑统计类型
			kafkaTemplate.send(KafkaConstant.QUEUE_ALBUM_STAT_UPDATE, JSONObject.toJSONString(albumStatMqVo));
		} else
		{
			log.info("获取锁失败,直接返回");
		}
	}

	@Override
	public Map<String, Object> getLatelyTrack() {
		HashMap<String, Object> resultMap = new HashMap<>();
		// todo:是通过MongoDB中存储声音记录去查询！{按照 userId, updateTime 排序 取出 albumId,trackId}
		UserListenProcess listenProcess = mongoTemplate.findOne(Query
						.query(Criteria
								.where("userId")
								.is(AuthContextHolder.getUserId()))
				//根据更新时间获取上一次播放声音
						.with(Sort.by(Sort.Direction.DESC, "updateTime"))
				//映射对象
				, UserListenProcess.class
				//集合名字
				, MongoUtil.getCollectionName(MongoUtil.MongoCollectionEnum.USER_LISTEN_PROCESS, AuthContextHolder.getUserId()));
		resultMap.put("trackId", listenProcess.getTrackId());
		resultMap.put("albumId", listenProcess.getAlbumId());
		return resultMap;
	}
}
